package com.shujia.spark.streaming

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo02UpdateStateByKey {
  def main(args: Array[String]): Unit = {

    // 状态：某个时刻程序运行的结果

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local[2]") // 至少需要两个线程，因为接收数据需要一致占用一个线程
      .config("spark.default.parallelism", "2")
      .getOrCreate()

    val ssc = new StreamingContext(spark.sparkContext, Durations.seconds(5))

    // 使用有状态算子时必须要指定CK的目录，状态会通过CK进行保存
    ssc.checkpoint("spark/data/ck")

    val lineDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    lineDS
      .flatMap(_.split(","))
      .map(w => (w, 1))
      .updateStateByKey((seq: Seq[Int], opt: Option[Int]) => {
        // seq：当前5s批次内接收到的数据按照Key分组后，某个Key所有的Value
        // 统计当前批次某个Key对应的单词数量
        val currentBatchRes: Int = seq.sum

        opt match {
          // 之前批次的结果
          case Some(lastBatchRes) =>
            Some(currentBatchRes + lastBatchRes)
          // 当前key是第一次被处理，之前没有状态
          case None =>
            Some(currentBatchRes)
        }
      }).print()

    ssc.start()
    ssc.awaitTermination()


  }

}
